package cn.com.client.send;

import cn.com.client.model.MergeKeyModel;
import cn.com.client.reference.ZhurongReference;
import cn.com.common.Constant;
import cn.com.common.Invocation;
import cn.com.common.RegistryInfo;
import cn.com.common.URI;
import cn.com.common.model.ProbeModel;
import cn.com.common.utils.IPHelper;
import cn.com.common.utils.SuspendUtils;
import cn.com.rpc.invoke.Invoker;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author jiaming
 */
public class KeySender implements Runnable {

    protected static final Logger log = LogManager.getLogger();

    private final RecordAccumulator recordAccumulator = new RecordAccumulator();

    public static volatile boolean SEND = true;
    /**
     * 最大获取记录数
     */
    private static final int MAX_RECORD_SIZE = 100;
    /**
     * 最大等待时间，单位ms
     */
    private static final int MAX_WAITING_TIME = 100;

    private ZhurongReference reference;

    private Invoker invoker;

    public KeySender() {
        init();
    }

    private void init() {
        this.reference = new ZhurongReference();
        RegistryInfo registryInfo = new RegistryInfo();
        registryInfo.setHost(Constant.ETCD_HOST);
        registryInfo.setPort(Constant.ETCD_PORT);
        registryInfo.setAppName(Constant.APP_NAME);
        URI uri = new URI();
        uri.setHost(IPHelper.getHostFirstIp());
        uri.setPort(Constant.CLIENT_PORT);
        uri.setServiceName(Constant.APP_NAME);
        this.invoker = reference.refer(registryInfo, uri);
    }

    @Override
    public void run() {
        while (SEND) {
            long startTimeMillis = System.currentTimeMillis();
            List<ProbeModel> results = new ArrayList<>(MAX_RECORD_SIZE);
            recordAccumulator.record(results, MAX_RECORD_SIZE);
            if (results.isEmpty()) {
                //睡眠1毫秒
                SuspendUtils.suspendInMILLISECONDS(1);
                continue;
            }
            while (true) {
                int remaining = (MAX_RECORD_SIZE - results.size());
                if ((System.currentTimeMillis() - startTimeMillis) > MAX_WAITING_TIME || remaining == 0) {
                    //time is end or results is full,send request
                    this.send(results);
                    break;
                } else {
                    List<ProbeModel> remainingResults = new ArrayList<>(remaining);
                    recordAccumulator.record(remainingResults, remaining);
                    results.addAll(remainingResults);
                }
            }
        }
    }

    private void send(List<ProbeModel> results) {
        if (results.isEmpty()) {
            return;
        }
        try {
            Map<String, MergeKeyModel> keySumMap = new HashMap<>();
            //汇总统计key发送
            for (ProbeModel result : results) {
                MergeKeyModel model = keySumMap.get(result.getKey());
                if (model == null) {
                    keySumMap.put(result.getKey(), new MergeKeyModel(result.getAppName(), 1));
                } else {
                    model.addSum(1);
                }
            }
            List<ProbeModel> probeModels = new ArrayList<>(keySumMap.size());
            for (Map.Entry<String, MergeKeyModel> entry : keySumMap.entrySet()) {
                String key = entry.getKey();
                MergeKeyModel mergeKeyModel = entry.getValue();
                ProbeModel probeModel = new ProbeModel(mergeKeyModel.getAppName(), 1, key, mergeKeyModel.getSum());
                probeModels.add(probeModel);
            }
            Invocation invocation = new Invocation("hotkey", new Object[]{probeModels});
            try {
                invoker.invoke(invocation);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }

    }

}